package com.example.status;

import com.example.bean.WaterSenSorFunction;
import com.example.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * Created with IntelliJ IDEA.
 * ClassName: KeyedValuesStatusDemo
 * Package: com.example.status
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-21
 * Time: 15:16
 */

//状态值 知识保存一个值
public class KeyedValuesStatusDemo {
    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //获取数据源
        final SingleOutputStreamOperator<WaterSensor> dataWT = env.socketTextStream("hadoop103", 7777)
                //转换算子 将获取的数据源封装为对象
                .map(new WaterSenSorFunction())
                //分配时间窗口设置水位线
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                //乱序流中水位线设置 参数表示 乱序的最大程是3秒
                                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                //时间提取提 从数据中提取事件事件
                                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                                    @Override
                                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                        //单位值1毫秒
                                        return element.getTs() + 1000L;
                                    }
                                })
                );

        //状态值 算子
        //检测每种传感器的水位值，如果连续的两个水位值超过10，就输出报警
        //根据id进行分组
        final SingleOutputStreamOperator<String> processStatus = dataWT.keyBy(r -> r.getId())
                //分析这个需求 没有可以直接使用的算子 那就用process
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    //值状态 定义状态
                    //虽然这里之定义了一个状态 但是不同的key维护着不同的状态
                    ValueState<Integer> lastVC;
                    //而且不能在这里初始化 和类的加载顺序有关 获取不到上下文
                    //如果这里使用普通变量创建状态 s1 , s2谁是谁的状态是无法维护的
                    //如果是普通变量 要有一个HashMap来维护 存一个相同的key来判断
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //2.在open初始化方法中定义状态
                        lastVC = getRuntimeContext().
                                //通过get获取运行时上下文 getStatus 获取状态
                                //参数是一个 new一个值状态描述器 参数是 起一个状态的名字 唯一不重复
                                //参数二 数据的类型
                                        getState(new ValueStateDescriptor<Integer>("vc", Types.INT));
                    }
                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                        //连续两个值 作比较 说明要拿到上一个值 要存起来

                        //lastVC.value() //取出之状态的数据
                        //lastVC.update(); //更新值状态的数据
                        //lastVC.clear(); //删除值状态的数据

                        //取数上一条数据的水位值
                        //包装类的默认值是null
                        int vc = lastVC.value() == null ? 0 : lastVC.value();

                        //新来的值 和 状态值 取绝对值 判断是否大于10
                        if (Math.abs(value.getVc() - vc) > 10) {
                            out.collect("传感器ID: " +value.getId()+" 当前水位值=" + value.getVc() + "与上一条水位值=" + vc + "相差大于十");
                        }
                        //求差值

                        //更新水位值

                        lastVC.update(value.getVc());

                    }
                });

        processStatus.print();


        env.execute();
    }
}
